package com.supplychain.scm.core.rocketmq.producer;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.supplychain.scm.core.basis.constant.StringPool;
import com.supplychain.scm.core.rocketmq.bean.BaseMqMessage;
import com.supplychain.scm.core.rocketmq.bean.MessageTypeEnum;
import com.supplychain.scm.core.rocketmq.constant.RocketMqConstant;
import com.supplychain.scm.core.rocketmq.function.AsyncFunction;
import com.supplychain.scm.core.rocketmq.storage.IMqMessageStorage;
import com.supplychain.scm.core.rocketmq.util.RocketThreadLocalUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.DelayMode;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author fengshaoyu
 * @version 1.0.0
 * <p></p>
 * @date 2022/5/27 16:46
 */
@Slf4j
@RequiredArgsConstructor
public class RocketMqProducer {

    protected final RocketMQTemplate rocketMQTemplate;
    protected final IMqMessageStorage mqMessageStorage;
    protected final Executor executor;
    protected final String applicationName;
    protected final Boolean ifStore;
    /**
     * 同步消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param messageLinkId 消息链路ID
     * @param <T>         class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload, MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag,uniqueKey, messageLinkId, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, build));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 同步消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param messageLinkId 消息链路ID
     * @param <T>         class
     */
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, String messageLinkId) {
        return this.syncSendMessage(topic,withPayload,tag, StringPool.EMPTY,messageLinkId);
    }

    /**
     * 同步消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param <T>         class
     */
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag) {
        return this.syncSendMessage(topic,withPayload,tag, StringPool.EMPTY);
    }

    /**
     * 同步消息
     *
     * @param topic        topic
     * @param withPayloads list<Payloads>
     * @param tag          tag
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendMessage(String topic, List<T> withPayloads, String tag, String uniqueKey, String messageLinkId) {
        Assert.isTrue(CollectionUtils.isNotEmpty(withPayloads) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        //将所有对象toString
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<String> messages = this.buildMessage(JSON.toJSONString(withPayloads),MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayloads, tag, uniqueKey, messageLinkId, null, null, null, null);
        SendResult sendResult =this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, messages));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayloads));
        }
        return sendStatus;
    }

    /**
     * 同步消息
     *
     * @param topic        topic
     * @param withPayloads list<Payloads>
     * @param tag          tag
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendMessage(String topic, List<T> withPayloads, String tag, String messageLinkId) {
        return this.syncSendMessage(topic, withPayloads, tag, StringPool.EMPTY,messageLinkId);
    }

    /**
     * 同步消息
     *
     * @param topic        topic
     * @param withPayloads list<Payloads>
     * @param tag          tag
     * @param <T>          class
     */
    public final <T> SendStatus syncSendMessage(String topic, List<T> withPayloads, String tag) {
        return this.syncSendMessage(topic, withPayloads, tag, StringPool.EMPTY);
    }

    /**
     * 同步延时消息（秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @param uniqueKey
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeSeconds(String topic, T withPayload, String tag, Long delayTime,String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDelayTimeSeconds(dest, build, delayTime));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 同步延时消息（毫秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @param uniqueKey
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeMills(String topic, T withPayload, String tag, Long delayTime,String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDelayTimeMills(dest, build, delayTime));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 同步定时消息（毫秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param deliverTimeMills
     * @param uniqueKey
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDeliverMessageTimeMills(String topic, T withPayload, String tag, Long deliverTimeMills,String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDeliverTimeMills(dest, build, deliverTimeMills));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 同步延迟消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param timeout     timeout
     * @param delayLevel  延迟级别
     * @param <T>         class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    @Deprecated
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, Long timeout, Integer delayLevel,String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, build, timeout, delayLevel));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 同步延迟消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param timeout     timeout
     * @param delayLevel  延迟级别
     * @param <T>         class
     * @param messageLinkId 消息链路ID
     */
    @Deprecated
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, Long timeout, Integer delayLevel, String messageLinkId) {
        return this.syncSendMessage(topic, withPayload, tag, timeout, delayLevel, null,messageLinkId);
    }

    @Deprecated
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, Integer delayLevel, String messageLinkId) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        return this.syncSendMessage(topic, withPayload, tag, Long.valueOf(sendMsgTimeout), delayLevel, null, messageLinkId);
    }

    @Deprecated
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, Integer delayLevel) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        return this.syncSendMessage(topic, withPayload, tag, Long.valueOf(sendMsgTimeout), delayLevel, null, null);
    }
    /**
     * 同步延时消息（秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeSeconds(String topic, T withPayload, String tag, Long delayTime, String messageLinkId) {
        return this.syncSendDelayMessageTimeSeconds(topic, withPayload, tag,  delayTime, null,messageLinkId);
    }

    /**
     * 同步延时消息（毫秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeMills(String topic, T withPayload, String tag, Long delayTime, String messageLinkId) {
        return this.syncSendDelayMessageTimeMills(topic, withPayload, tag,  delayTime, null,messageLinkId);
    }

    /**
     * 同步定时消息（秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param deliverTimeMills
     * @param messageLinkId
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDeliverMessageTimeMills(String topic, T withPayload, String tag, Long deliverTimeMills, String messageLinkId) {
        return this.syncSendDeliverMessageTimeMills(topic, withPayload, tag,  deliverTimeMills, null,messageLinkId);
    }

    /**
     * 同步延迟消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param timeout     timeout
     * @param delayLevel  延迟级别
     * @param <T>         class
     */
    @Deprecated
    public final <T> SendStatus syncSendMessage(String topic, T withPayload, String tag, Long timeout, Integer delayLevel) {
        return this.syncSendMessage(topic, withPayload, tag, timeout, delayLevel, null);
    }

    /**
     * 同步延时消息（秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeSeconds(String topic, T withPayload, String tag, Long delayTime) {
        return this.syncSendDelayMessageTimeSeconds(topic, withPayload, tag,  delayTime, null,null);
    }

    /**
     * 同步延时消息（毫秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param delayTime
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDelayMessageTimeMills(String topic, T withPayload, String tag, Long delayTime) {
        return this.syncSendDelayMessageTimeMills(topic, withPayload, tag,  delayTime, null,null);
    }

    /**
     * 同步定时消息（秒）
     * @param topic
     * @param withPayload
     * @param tag
     * @param deliverTimeMills
     * @return
     * @param <T>
     */
    public final <T> SendStatus syncSendDeliverMessageTimeMills(String topic, T withPayload, String tag, Long deliverTimeMills) {
        return this.syncSendDeliverMessageTimeMills(topic, withPayload, tag,  deliverTimeMills, null,null);
    }

    /**
     * 同步消息
     *
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    @Deprecated
    public <T> SendStatus syncSendMessage(String topic, String tag, Message<T> message,Long timeout, Integer delayLevel) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, timeout, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, message,timeout,delayLevel));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 同步延时消息（秒）
     * @param topic
     * @param tag
     * @param message
     * @param delayTime
     * @return
     * @param <T>
     */
    public <T> SendStatus syncSendDelayMessageTimeSeconds(String topic, String tag, Message<T> message, Long delayTime) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDelayTimeSeconds(dest, message,delayTime));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 同步延时消息（毫秒）
     * @param topic
     * @param tag
     * @param message
     * @param delayTime
     * @return
     * @param <T>
     */
    public <T> SendStatus syncSendDelayMessageTimeMills(String topic, String tag, Message<T> message, Long delayTime) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDelayTimeMills(dest, message,delayTime));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 同步定时消息（秒）
     * @param topic
     * @param tag
     * @param message
     * @param deliverTimeMills
     * @return
     * @param <T>
     */
    public <T> SendStatus syncSendDeliverMessageTimeMills(String topic, String tag, Message<T> message, Long deliverTimeMills) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendDeliverTimeMills(dest, message,deliverTimeMills));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 同步消息
     *
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    public <T> SendStatus syncSendMessage(String topic, String tag, Message<T> message,Long timeout) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, timeout, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, message,timeout));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 同步消息
     *
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    public <T> SendStatus syncSendMessage(String topic, String tag, Message<T> message) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, message, tag, null,null, null, null, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSend(dest, message));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(message));
        }
        return sendStatus;
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param hashKey     hashKey
     * @param <T>         class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, hashKey, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendOrderly(dest, build, hashKey));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return  sendStatus;
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param hashKey     hashKey
     * @param <T>         class
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, String hashKey, String messageLinkId) {
        return this.syncSendOrderlyMessage(topic, withPayload, tag, hashKey,null,messageLinkId);
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload Payload
     * @param tag         tag
     * @param hashKey     hashKey
     * @param <T>         class
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, String hashKey) {
        return this.syncSendOrderlyMessage(topic, withPayload, tag, hashKey,null);
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload payload
     * @param tag         tag
     * @param timeout     timeout
     * @param hashKey     hashKey
     * @param <T>         class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, Long timeout, String hashKey,String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toSyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, hashKey, null, null);
        SendResult sendResult = this.syncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.syncSendOrderly(dest, build, hashKey, timeout));
        SendStatus sendStatus = Optional.ofNullable(sendResult).map(SendResult::getSendStatus).orElse(null);
        if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
            log.error("Mq消息发送异常 => topic:{} tag:{} data:{}",topic,tag,JSON.toJSONString(withPayload));
        }
        return sendStatus;
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload payload
     * @param tag         tag
     * @param timeout     timeout
     * @param hashKey     hashKey
     * @param <T>         class
     * @param messageLinkId 消息链路ID
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, Long timeout, String hashKey, String messageLinkId) {
        return this.syncSendOrderlyMessage(topic, withPayload, tag, timeout, hashKey, null,messageLinkId);
    }

    /**
     * 异步顺序消息
     *
     * @param topic       topic
     * @param withPayload payload
     * @param tag         tag
     * @param timeout     timeout
     * @param hashKey     hashKey
     * @param <T>         class
     */
    public final <T> SendStatus syncSendOrderlyMessage(String topic, T withPayload, String tag, Long timeout, String hashKey) {
        return this.syncSendOrderlyMessage(topic, withPayload, tag, timeout, hashKey, null);
    }

    /**
     * 异步消息
     *int delayLevel
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    public <T> void asyncSendMessage(String topic, String tag, Message<T> message,Long timeout,int delayLevel) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, message, tag, null,null, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.asyncSend(dest, message,new RocketMqProducer.MsgLogSendCallback(baseMqMessage),timeout,delayLevel));
    }

    /**
     * 异步消息
     *
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    public <T> void asyncSendMessage(String topic, String tag, Message<T> message,Long timeout) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, message, tag, null,null, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.asyncSend(dest, message,new RocketMqProducer.MsgLogSendCallback(baseMqMessage),timeout));
    }

    /**
     * 异步消息
     *
     * @param topic      topic
     * @param tag        tag
     * @param message<T> 自定义message
     */
    public <T> void asyncSendMessage(String topic, String tag, Message<T> message) {
        Assert.isTrue(Objects.nonNull(message), "message, or tag is null");
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, message, tag, null,null, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage, () -> rocketMQTemplate.asyncSend(dest, message,new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, build, sendCallback));
    }
    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, String messageLinkId) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayload,null,messageLinkId);
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param <T>          class
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayload, StringPool.EMPTY);
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads, String uniqueKey, String messageLinkId) {
        Assert.isTrue(CollectionUtils.isNotEmpty(withPayloads) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        String finalMessageLinkId = messageLinkId;
        String finalUniqueKey = uniqueKey;
        List<Message<T>> messages = withPayloads.stream().map(withPayload -> this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE, finalMessageLinkId, finalUniqueKey)).collect(Collectors.toList());
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayloads, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, messages, sendCallback));
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads, String messageLinkId) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayloads, null,messageLinkId);
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param <T>          class
     */
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayloads, StringPool.EMPTY);
    }

    /**
     * 异步发送延时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param uniqueKey
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, sendCallback, timeout, delayTime,DelayMode.DELAY_SECONDS));
    }

    /**
     * 异步发送延时消息（毫秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param uniqueKey
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, sendCallback, timeout, delayTime,DelayMode.DELAY_MILLISECONDS));
    }

    /**
     * 异步发送定时消息（毫秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param deliverTimeMills
     * @param uniqueKey
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long deliverTimeMills, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, sendCallback, timeout, deliverTimeMills,DelayMode.DELIVER_TIME_MILLISECONDS));
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param delayLevel   延迟级别
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Integer delayLevel, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, build, sendCallback, timeout, delayLevel));
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param delayLevel   延迟级别
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Integer delayLevel, String messageLinkId) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayload, timeout, delayLevel, null,messageLinkId);
    }

    /**
     * 异步延时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime, String messageLinkId) {
        this.asyncSendDelayMessageTimeSeconds(topic, tag, sendCallback, withPayload, timeout, delayTime, null,messageLinkId);
    }

    /**
     * 异步延时消息（毫秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime, String messageLinkId) {
        this.asyncSendDelayMessageTimeMills(topic, tag, sendCallback, withPayload, timeout, delayTime, null,messageLinkId);
    }

    /**
     * 异步定时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param deliverTimeMills
     * @param messageLinkId
     * @param <T>
     */
    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long deliverTimeMills, String messageLinkId) {
        this.asyncSendDeliverMessageTimeMills(topic, tag, sendCallback, withPayload, timeout, deliverTimeMills, null,messageLinkId);
    }

    /**
     * 异步消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param delayLevel   延迟级别
     * @param <T>          class
     */
    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Integer delayLevel) {
        this.asyncSendMessage(topic, tag, sendCallback, withPayload, timeout, delayLevel, null);
    }

    /**
     * 异步延时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime) {
        this.asyncSendDelayMessageTimeSeconds(topic, tag, sendCallback, withPayload, timeout, delayTime, null,null);
    }

    /**
     * 异步延时消息（毫秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param delayTime
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long delayTime) {
        this.asyncSendDelayMessageTimeMills(topic, tag, sendCallback, withPayload, timeout, delayTime, null,null);
    }

    /**
     * 异步定时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param timeout
     * @param deliverTimeMills
     * @param <T>
     */
    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, Long deliverTimeMills) {
        this.asyncSendDeliverMessageTimeMills(topic, tag, sendCallback, withPayload, timeout, deliverTimeMills, null,null);
    }

    /**
     * 异步延时消息
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param delayLevel
     * @param <T>
     */
    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Integer delayLevel) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendMessage(topic, tag, sendCallback, withPayload, Long.valueOf(sendMsgTimeout), delayLevel, null);
    }

    /**
     * 异步延时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param delayTime
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, SendCallback sendCallback, T withPayload, Long delayTime) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDelayMessageTimeSeconds(topic, tag, sendCallback, withPayload, Long.valueOf(sendMsgTimeout), delayTime, null,null);
    }

    /**
     * 异步延时消息（毫秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param delayTime
     * @param <T>
     */
    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long delayTime) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDelayMessageTimeMills(topic, tag, sendCallback, withPayload, Long.valueOf(sendMsgTimeout), delayTime, null,null);
    }

    /**
     * 异步定时消息（秒）
     * @param topic
     * @param tag
     * @param sendCallback
     * @param withPayload
     * @param deliverTimeMills
     * @param <T>
     */
    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, SendCallback sendCallback, T withPayload, Long deliverTimeMills) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDeliverMessageTimeMills(topic, tag, sendCallback, withPayload, Long.valueOf(sendMsgTimeout), deliverTimeMills, null,null);
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param hashKey      hashKey
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, hashKey, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSendOrderly(dest, build, hashKey, sendCallback));
    }


    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param hashKey      hashKey
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, String hashKey, String messageLinkId) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayload, hashKey,null,messageLinkId);
    }


    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param hashKey      hashKey
     * @param <T>          class
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, String hashKey) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayload, hashKey,null);
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param hashKey      hashKey
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(CollectionUtils.isNotEmpty(withPayloads) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        String finalMessageLinkId = messageLinkId;
        String finalUniqueKey = uniqueKey;
        List<Message<T>> messages = withPayloads.stream().map(withPayload -> this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE, finalMessageLinkId, finalUniqueKey)).collect(Collectors.toList());
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayloads, tag, uniqueKey, messageLinkId, null, hashKey, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSendOrderly(dest, messages, hashKey, sendCallback));
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param hashKey      hashKey
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads, String hashKey, String messageLinkId) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayloads, hashKey,null,messageLinkId);
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayloads payload
     * @param hashKey      hashKey
     * @param <T>          class
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, List<T> withPayloads, String hashKey) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayloads, hashKey,null);
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param hashKey      hashKey
     * @param <T>          class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, hashKey, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSendOrderly(dest, build, hashKey, sendCallback, timeout));
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param hashKey      hashKey
     * @param <T>          class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, String hashKey, String messageLinkId) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayload, timeout, hashKey, null,messageLinkId);
    }

    /**
     * 异步顺序消息
     * 无数据落库处理
     *
     * @param topic        topic
     * @param tag          tag
     * @param sendCallback 回调接口
     * @param withPayload  payload
     * @param timeout      timeout
     * @param hashKey      hashKey
     * @param <T>          class
     */
    public final <T> void asyncSendOrderlyMessage(String topic, String tag, SendCallback sendCallback, T withPayload, Long timeout, String hashKey) {
        this.asyncSendOrderlyMessage(topic, tag, sendCallback, withPayload, timeout, hashKey, null);
    }

    /**
     * 异步消息
     * @param topic topic
     * @param tag tag
     * @param withPayload payload
     * @param <T> class
     * @param uniqueKey 唯一Key
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, build, new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    /**
     * 异步消息
     * @param topic topic
     * @param tag tag
     * @param withPayload payload
     * @param <T> class
     * @param messageLinkId 消息链路ID
     */
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, String messageLinkId) {
        this.asyncSendMessage(topic, tag, withPayload,null,messageLinkId);
    }

    /**
     * 异步消息
     * @param topic topic
     * @param tag tag
     * @param withPayload payload
     * @param <T> class
     */
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload) {
        this.asyncSendMessage(topic, tag, withPayload,StringPool.EMPTY);
    }

    public final <T> void asyncSendMessage(String topic, String tag, List<T> withPayloads, String uniqueKey, String messageLinkId) {
        Assert.isTrue(CollectionUtils.isNotEmpty(withPayloads) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        String finalMessageLinkId = messageLinkId;
        String finalUniqueKey = uniqueKey;
        List<Message<T>> messages = withPayloads.stream().map(withPayload -> this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE, finalMessageLinkId, finalUniqueKey)).collect(Collectors.toList());
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayloads, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, messages, new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    public final <T> void asyncSendMessage(String topic, String tag, List<T> withPayloads, String messageLinkId) {
        this.asyncSendMessage(topic, tag, withPayloads, null,messageLinkId);
    }

    public final <T> void asyncSendMessage(String topic, String tag, List<T> withPayloads) {
        this.asyncSendMessage(topic, tag, withPayloads, StringPool.EMPTY);
    }

    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, Long timeout, Integer delayLevel, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSend(dest, build, new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, T withPayload, Long timeout, Long delayTime, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, new RocketMqProducer.MsgLogSendCallback(baseMqMessage),timeout,delayTime,DelayMode.DELAY_SECONDS));
    }

    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long delayTime, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, new RocketMqProducer.MsgLogSendCallback(baseMqMessage),timeout,delayTime,DelayMode.DELAY_MILLISECONDS));
    }

    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long deliverTimeMills, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.UNORDERED_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, null, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> this.asyncSend(dest, build, new RocketMqProducer.MsgLogSendCallback(baseMqMessage),timeout,deliverTimeMills,DelayMode.DELIVER_TIME_MILLISECONDS));
    }

    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, Long timeout, Integer delayLevel, String messageLinkId) {
        this.asyncSendMessage(topic, tag, withPayload, timeout, delayLevel,null,messageLinkId);
    }

    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, T withPayload, Long timeout, Long delayTime, String messageLinkId) {
        this.asyncSendDelayMessageTimeSeconds(topic, tag, withPayload, timeout, delayTime,null,messageLinkId);
    }

    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long delayTime, String messageLinkId) {
        this.asyncSendDelayMessageTimeMills(topic, tag, withPayload, timeout, delayTime,null,messageLinkId);
    }

    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long deliverTimeMills, String messageLinkId) {
        this.asyncSendDeliverMessageTimeMills(topic, tag, withPayload, timeout, deliverTimeMills,null,messageLinkId);
    }

    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, Long timeout, Integer delayLevel) {
        this.asyncSendMessage(topic, tag, withPayload, timeout, delayLevel,null);
    }

    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, T withPayload, Long timeout, Long delayTime) {
        this.asyncSendDelayMessageTimeSeconds(topic, tag, withPayload, timeout, delayTime,null,null);
    }

    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long delayTime) {
        this.asyncSendDelayMessageTimeMills(topic, tag, withPayload, timeout, delayTime,null,null);
    }

    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, T withPayload, Long timeout, Long deliverTimeMills) {
        this.asyncSendDeliverMessageTimeMills(topic, tag, withPayload, timeout, deliverTimeMills,null,null);
    }

    @Deprecated
    public final <T> void asyncSendMessage(String topic, String tag, T withPayload, Integer delayLevel) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendMessage(topic, tag, withPayload, Long.valueOf(sendMsgTimeout), delayLevel,null);
    }

    public final <T> void asyncSendDelayMessageTimeSeconds(String topic, String tag, T withPayload,Long delayTime) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDelayMessageTimeSeconds(topic, tag, withPayload, Long.valueOf(sendMsgTimeout), delayTime,null,null);
    }

    public final <T> void asyncSendDelayMessageTimeMills(String topic, String tag, T withPayload, Long delayTime) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDelayMessageTimeMills(topic, tag, withPayload, Long.valueOf(sendMsgTimeout), delayTime,null,null);
    }

    public final <T> void asyncSendDeliverMessageTimeMills(String topic, String tag, T withPayload, Long deliverTimeMills) {
        int sendMsgTimeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
        this.asyncSendDeliverMessageTimeMills(topic, tag, withPayload, Long.valueOf(sendMsgTimeout), deliverTimeMills,null,null);
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, null, hashKey, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSendOrderly(dest, build, hashKey, new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, String hashKey, String messageLinkId) {
        this.asyncSendOrderlyMessage(topic, tag, withPayload, hashKey,null,messageLinkId);
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, String hashKey) {
        this.asyncSendOrderlyMessage(topic, tag, withPayload, hashKey,null);
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, Long timeout, String hashKey, String uniqueKey, String messageLinkId) {
        Assert.isTrue(Objects.nonNull(withPayload) || StringUtils.isNotBlank(tag), "withPayload, or tag is null");
        if(StringUtils.isBlank(uniqueKey)) {
            uniqueKey = String.valueOf(IdUtil.getSnowflakeNextId());
        }
        String consumerMessageLinkId = RocketThreadLocalUtils.getMessageLinkId();
        if(StringUtils.isNotBlank(consumerMessageLinkId)) {
            messageLinkId = consumerMessageLinkId;
        }
        if(StringUtils.isBlank(messageLinkId)) {
            messageLinkId = uniqueKey;
        }
        Message<T> build = this.buildMessage(withPayload,MessageTypeEnum.ORDERLY_MESSAGE,messageLinkId,uniqueKey);
        String dest = String.format(RocketMqConstant.FORMAT, topic, tag);
        BaseMqMessage baseMqMessage = this.toAsyncBaseMqMessage(topic, withPayload, tag, uniqueKey, messageLinkId, timeout, hashKey, null, null);
        this.asyncSendMessageFunction(baseMqMessage,() -> rocketMQTemplate.asyncSendOrderly(dest, build, hashKey, new RocketMqProducer.MsgLogSendCallback(baseMqMessage)));
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, Long timeout, String hashKey, String messageLinkId) {
        this.asyncSendOrderlyMessage(topic, tag, withPayload, timeout, hashKey,null,messageLinkId);
    }

    public final <T> void asyncSendOrderlyMessage(String topic, String tag, T withPayload, Long timeout, String hashKey) {
        this.asyncSendOrderlyMessage(topic, tag, withPayload, timeout, hashKey,null);
    }

    private void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout, long delayTime, DelayMode mode) {
        if (Objects.nonNull(message) && Objects.nonNull(message.getPayload())) {
            try {
                long now = System.currentTimeMillis();
                org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);
                if (delayTime > 0L && Objects.nonNull(mode)) {
                    switch (mode) {
                        case DELAY_SECONDS:
                            rocketMsg.setDelayTimeSec(delayTime);
                            break;
                        case DELAY_MILLISECONDS:
                            rocketMsg.setDelayTimeMs(delayTime);
                            break;
                        case DELIVER_TIME_MILLISECONDS:
                            rocketMsg.setDeliverTimeMs(delayTime);
                            break;
                        default:
                            log.warn("delay mode: {} not support", mode);
                    }
                }
                DefaultMQProducer producer = rocketMQTemplate.getProducer();
                producer.send(rocketMsg, sendCallback,timeout);
                long costTime = System.currentTimeMillis() - now;
                if (log.isDebugEnabled()) {
                    log.debug("send message cost: {} ms", costTime);
                }
            } catch (Exception var14) {
                log.error("syncSend failed. destination:{}, message:{}, detail exception info: ", new Object[]{destination, message, var14});
                throw new MessagingException(var14.getMessage(), var14);
            }
        } else {
            log.error("syncSend failed. destination:{}, message is null ", destination);
            throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
        }
    }

    private org.apache.rocketmq.common.message.Message createRocketMqMessage(String destination, Message<?> message) {
        Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), (MessagePostProcessor)null);
        return RocketMQUtil.convertToRocketMessage(rocketMQTemplate.getMessageConverter(), rocketMQTemplate.getCharset(), destination, msg);
    }

    private Message<?> doConvert(Object payload, Map<String, Object> headers, MessagePostProcessor postProcessor) {
        Message<?> message = this.doConvertMessage(payload, headers, postProcessor);
        MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
        builder.setHeaderIfAbsent("contentType", MimeTypeUtils.TEXT_PLAIN);
        return builder.build();
    }

    private Message<?> doConvertMessage(Object payload, @Nullable Map<String, Object> headers, @Nullable MessagePostProcessor postProcessor) {
        MessageHeaders messageHeaders = null;
        Object conversionHint = Objects.nonNull(headers) ? headers.get("conversionHint") : null;
        Map<String, Object> headersToUse = this.processHeadersToSend(headers);
        if (Objects.nonNull(headersToUse)) {
            if (headersToUse instanceof MessageHeaders) {
                messageHeaders = (MessageHeaders)headersToUse;
            } else {
                messageHeaders = new MessageHeaders(headersToUse);
            }
        }

        MessageConverter converter = rocketMQTemplate.getMessageConverter();
        Message<?> message = converter instanceof SmartMessageConverter ? ((SmartMessageConverter)converter).toMessage(payload, messageHeaders, conversionHint) : converter.toMessage(payload, messageHeaders);
        if (message == null) {
            String payloadType = payload.getClass().getName();
            Object contentType = Objects.nonNull(messageHeaders) ? messageHeaders.get("contentType") : null;
            throw new MessageConversionException("Unable to convert payload with type='" + payloadType + "', contentType='" + contentType + "', converter=[" + rocketMQTemplate.getMessageConverter() + "]");
        } else {
            if (Objects.nonNull(postProcessor)) {
                message = postProcessor.postProcessMessage(message);
            }

            return message;
        }
    }

    @Nullable
    private Map<String, Object> processHeadersToSend(@Nullable Map<String, Object> headers) {
        return headers;
    }

    private BaseMqMessage toSyncBaseMqMessage(String topic, Object withPayload, String tag,
                                          String uniqueKey,String messageLinkId, Long timeout,
                                          String hashKey, String msgId, String sendStatus){
            return this.buildBaseMqMessage(this.notSaveSupplier(topic, tag),Boolean.TRUE,uniqueKey, messageLinkId, topic, withPayload, tag, timeout, hashKey, msgId, sendStatus);

    }

    private BaseMqMessage toAsyncBaseMqMessage(String topic, Object withPayload, String tag,
                                               String uniqueKey,String messageLinkId, Long timeout,
                                               String hashKey, String msgId, String sendStatus){
        return this.buildBaseMqMessage(this.notSaveSupplier(topic, tag),Boolean.FALSE,uniqueKey, messageLinkId, topic, withPayload, tag, timeout, hashKey, msgId, sendStatus);
    }

    private void createMessage(BaseMqMessage baseMqMessage,Executor executor) {
        if(log.isDebugEnabled()) {
            log.debug("mq send create message = {}",JSON.toJSONString(baseMqMessage));
        }
        if(Objects.nonNull(mqMessageStorage)) {
            this.mqMessageStorage.saveMessage(baseMqMessage,executor);
        }
    }

    private <T extends SendResult> T syncSendMessageFunction(BaseMqMessage baseMqMessage, Supplier<T> supplier) {
        try {
            T t = supplier.get();
            baseMqMessage.setStatus(t.getSendStatus().name());
            baseMqMessage.setMsgId(t.getMsgId());
            return t;
        } catch (Throwable e) {
            baseMqMessage.setStatus(RocketMqConstant.FAIL);
            baseMqMessage.setError(ExceptionUtils.getStackTrace(e));
            log.error("Mq消息发送异常 ex => {}", ExceptionUtils.getStackTrace(e));
        } finally {
            this.createMessage(baseMqMessage,executor);
        }
        return null;
    }

    private void asyncSendMessageFunction(BaseMqMessage baseMqMessage, AsyncFunction function) {
        baseMqMessage.setStatus(RocketMqConstant.SENDING);
        this.createMessage(baseMqMessage,executor);
        try {
            function.accept();
        } catch (Throwable e) {
            baseMqMessage.setStatus(RocketMqConstant.FAIL);
            baseMqMessage.setError(ExceptionUtils.getStackTrace(e));
            log.error("Mq消息发送异常 ex => {}", ExceptionUtils.getStackTrace(e));
        } /*finally {
            baseMqMessage.setStatus(RocketMqConstant.SENDING);
            this.createMessage(baseMqMessage,executor);
        }*/
    }

    public <T> Message<T> buildMessage(T withPayload, MessageTypeEnum messageTypeEnum, String messageLinkId, String uniqueKey){
        return MessageBuilder
                .withPayload(withPayload)
                .setHeader(RocketMqConstant.MESSAGE_LINK_ID,messageLinkId)
                .setHeader(RocketMqConstant.KEYS,uniqueKey)
                .setHeader(RocketMqConstant.MESSAGE_TYPE,messageTypeEnum.getValue())
                .setHeader(RocketMqConstant.MESSAGE_ORIGIN,applicationName)
                .build();
    }
//
//    /**
//     * 默认回调
//     */
//    public class DefaultSendCallback implements SendCallback {
//
//        @Override
//        public void onSuccess(SendResult sendResult) {
//            log.info("RocketMq 消息发送成功 sendResult = {}", JSON.toJSON(sendResult));
//        }
//
//        @Override
//        public void onException(Throwable throwable) {
//            log.warn("RocketMq 消息发送失败 errorMessage = {}", throwable.getMessage());
//        }
//    }

    public class MsgLogSendCallback implements SendCallback {

        private BaseMqMessage baseMqMessage;

        public MsgLogSendCallback(BaseMqMessage baseMqMessage) {
            this.baseMqMessage = baseMqMessage;
        }

        @Override
        public void onSuccess(SendResult sendResult) {
            String msgId = sendResult.getMsgId();
            baseMqMessage.setMsgId(msgId);
            baseMqMessage.setStatus(sendResult.getSendStatus().name());
            baseMqMessage.setIfStore(ifStore);
            if(log.isDebugEnabled()) {
                log.debug("RocketMq 异步消息发送成功 baseMqMessage = {},sendResult = {}",JSON.toJSONString(baseMqMessage), JSON.toJSONString(sendResult));
            }
            if (Objects.nonNull(mqMessageStorage)) {
                mqMessageStorage.updateMessage(baseMqMessage, executor);
            }
        }

        @Override
        public void onException(Throwable throwable) {
            baseMqMessage.setStatus(RocketMqConstant.FAIL);
            baseMqMessage.setError(ExceptionUtils.getStackTrace(throwable));
            baseMqMessage.setIfStore(ifStore);
            log.error("RocketMq 异步消息发送失败 baseMqMessage = {},errorMessage = {}",JSON.toJSONString(baseMqMessage), ExceptionUtils.getStackTrace(throwable));
            if (Objects.nonNull(mqMessageStorage)) {
                mqMessageStorage.updateMessage(baseMqMessage, executor);
            }
        }
    }

    /**
     * 是否需要落库
     *
     * @param topic topic
     * @param tag   tag
     * @return bool
     */
    protected boolean notSaveSupplier(String topic, String tag) {
        if(RocketMqConstant.EXCLUDE_TOPICS.contains(topic) || RocketMqConstant.EXCLUDE_TAGS.contains(tag)){
            return true;
        } else {
            return false;
        }
    }

    private BaseMqMessage buildBaseMqMessage(Boolean notSave,Boolean ifSync,String uniqueKey,String messageLinkId, String topic,
                                             Object withPayload, String tag, Long timeout,
                                             String hashKey, String msgId, String sendStatus){
        return BaseMqMessage.builder()
                .notSave(notSave)
                .ifSync(ifSync)
                .message(JSON.toJSONString(withPayload))
                .tag(tag)
                .status(sendStatus)
                .topic(topic)
                .timeout(timeout)
                .hashKey(hashKey)
                .msgId(msgId)
                .uniqueKey(uniqueKey)
                .messageLinkId(messageLinkId)
                .origin(applicationName)
                .ifStore(ifStore)
                .build();
    }
}
